//
// Created by DELL on 2025/2/5.
//

#include "ThreadPool.h"

#include <iostream>
#include <thread>

namespace zdotp {
    int Thread::generate_id = 0;
    constexpr int THREAD_MAX_THRESHOLD = 8;
    constexpr int TASK_MAX_THRESHOLD = 1024;
    constexpr int THREAD_IDLE_TIME = 2; // 空闲时间, 就回收线程
    ThreadPool::ThreadPool()
    : mode_(thread_pool_mode::MODE_FIXED),
    init_thread_num_(4),
    thread_size_threshold_(THREAD_MAX_THRESHOLD),
    current_thread_size_(0),
    idle_thread_size_(0),
    task_queue_max_threshold_(TASK_MAX_THRESHOLD),
    is_pool_running_(false)
    {}

    ThreadPool::~ThreadPool() {
        is_pool_running_ = false;

        std::unique_lock lock(task_queue_mutex_);
        // 抢锁 通知not_empty赶紧把is_pool_running过掉
        not_empty_.notify_all();
        exit_.wait(lock, [&] {
           return threads_.empty();
        });
    }

    void ThreadPool::set_mode(const thread_pool_mode mode) {
        if (check_thread_running()) {
            std::cout << "pool is running, cannot set_mode\n";
            return;
        }
        mode_ = mode;
    }

    void ThreadPool::set_task_queue_max_threshold(const unsigned int max_threshold) {
        if (check_thread_running()) {
            std::cout << "pool is running, cannot set_task_queue_max_threshold\n";
            return;
        }
        task_queue_max_threshold_ = max_threshold;
    }

    void ThreadPool::set_thread_size_threshold(int thread_threshold) {
        if (check_thread_running()) {
            std::cout << "pool is running, cannot set_thread_size_threshold\n";
            return;
        }
        if (mode_ == thread_pool_mode::MODE_CACHED) {
            thread_size_threshold_ = thread_threshold;
        }
    }

    void ThreadPool::start(const int init_thread_num) {
        if (check_thread_running()) {
            std::cout << "pool is running...\n";
            return;
        }

        is_pool_running_ = true;
        init_thread_num_ = init_thread_num;

        for (size_t i = 0; i < init_thread_num_; ++i) {
            add_and_start();
        }
    }

    Result ThreadPool::submit_task(std::shared_ptr<TaskBase> task) {
        if (!check_thread_running()) {
            std::cout << "pool is running, cannot submit_task\n";
            return {task, false};
        }
        std::unique_lock lock(task_queue_mutex_);
        // 等待到任务队列不满才能继续执行, 返回false任务超时
        if (!not_full_.wait_for(lock, std::chrono::seconds(1),
            [this] { return task_size_ < task_queue_max_threshold_; })) {
            std::cout << "task queue is full, task will be dropped" << std::endl;
            return {task, false};
        }

        task_queue_.emplace(task);
        ++task_size_;

        if (mode_ == thread_pool_mode::MODE_CACHED
            && task_size_ > idle_thread_size_
            && current_thread_size_ < thread_size_threshold_) {
            auto u_ptr = std::make_unique<Thread>([this](const int id) {
                thread_func(id);
            });
            int id = u_ptr->get_thread_id();
            threads_.emplace(id, std::move(u_ptr));
            // 不能使用已经move的值 调试出结果
            threads_[id]->start();
            ++current_thread_size_;
            ++idle_thread_size_;
            std::cout << "新增线程\n";

        }

        not_empty_.notify_all();

        return {task, true};
    }
    void ThreadPool::thread_func(const int thread_id) {
        auto last_time = std::chrono::system_clock::now();
        while (is_pool_running_) {
            std::shared_ptr<TaskBase> task;
            {
                std::unique_lock lock(task_queue_mutex_);

                while (task_size_ == 0) {
                    if (mode_ == thread_pool_mode::MODE_CACHED) {
                        if (std::cv_status::timeout ==
                            not_empty_.wait_for(lock, std::chrono::seconds(1))) {
                                auto now =std::chrono::high_resolution_clock::now();
                                const auto dur =
                                    std::chrono::duration_cast<std::chrono::seconds>(now - last_time).count();
                                if (dur > THREAD_IDLE_TIME && current_thread_size_ > init_thread_num_) {
                                    remove_and_exit(thread_id);
                                    return;
                                }
                            }
                    }else {
                        not_empty_.wait(lock);
                    }

                    if (!is_pool_running_) {
                        remove_and_exit(thread_id);
                        return;
                    }

                }

                --idle_thread_size_;

                task = task_queue_.front();
                task_queue_.pop();
                -- task_size_;

                if (task_size_ > 0) {
                    not_empty_.notify_all();
                }

                not_full_.notify_all();
            }
            if (task != nullptr) {
                task->exec();
            }

            last_time = std::chrono::system_clock::now();
            ++idle_thread_size_;
        }

        remove_and_exit(thread_id);
    }

    bool ThreadPool::check_thread_running() const {
        if (is_pool_running_) {
            return true;
        }
        return false;
    }

    void ThreadPool::add_and_start() {
        auto u_ptr = std::make_unique<Thread>([this](const int id) {
            thread_func(id);
        });
        int id = u_ptr->get_thread_id();
        threads_.emplace(id, std::move(u_ptr));
        ++idle_thread_size_; // 空闲线程个数
        ++current_thread_size_;
        threads_[id]->start();
        std::cout << "新增线程\n";

    }

    void ThreadPool::remove_and_exit(int thread_id) {
        --current_thread_size_;
        --idle_thread_size_;
        threads_.erase(thread_id);
        exit_.notify_all();
        std::cout << "减少线程\n";
    }
} // zdotp